package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo10Join {
  def main(args: Array[String]): Unit = {
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo10Join")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    // 1、读取students、scores数据
    val stuRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt")
    val scoRDD: RDD[String] = sc.textFile("Spark/data/stu/score.txt")

    // 2、将数据变成K-V格式
    // Key：关联的字段 id Value：将自身作为value

    val stuKVRDD: RDD[(String, String)] = stuRDD.map(line => (line.split(",")(0), line.replace(",", "|")))
    val scoKVRDD: RDD[(String, String)] = scoRDD.map(line => (line.split(",")(0), line.replace(",", "|")))

    // (1500100113,(1500100113|羿思真|24|女|理科三班,1500100113|1000001|96))
    // join只能作用在 K-V格式的RDD上，会按照K进行关联
    // join等同于inner join
    val joinRDD: RDD[(String, (String, String))] = stuKVRDD.join(scoKVRDD)

    joinRDD.map {
      case (id: String, (stu: String, sco: String)) =>
        // 使用 | 进行切分时需要转义
        val stuSplits: Array[String] = stu.split("\\|")
        val name: String = stuSplits(1)
        val clazz: String = stuSplits(4)
        val scoSplits: Array[String] = sco.split("\\|")
        val sid: String = scoSplits(1)
        val score: String = scoSplits(2)
        s"$id,$name,$clazz,$sid,$score"
    }.foreach(println)

    joinRDD.map(kv => {
      val id: String = kv._1
      val stuScoT2: (String, String) = kv._2
      val stu: String = stuScoT2._1
      val sco: String = stuScoT2._2

      val stuSplits: Array[String] = stu.split("\\|")
      val name: String = stuSplits(1)
      val clazz: String = stuSplits(4)
      val scoSplits: Array[String] = sco.split("\\|")
      val sid: String = scoSplits(1)
      val score: String = scoSplits(2)
      s"$id,$name,$clazz,$sid,$score"
    }).foreach(println)

    stuKVRDD
      .leftOuterJoin(scoKVRDD)
      .map {
        case (id: String, (stu: String, Some(sco: String))) =>
          val stuSplits: Array[String] = stu.split("\\|")
          val name: String = stuSplits(1)
          val clazz: String = stuSplits(4)
          val score: String = sco.split("\\|")(2)
          s"$id,$name,$clazz,$score"
        case (id: String, (stu: String, None)) =>
          val stuSplits: Array[String] = stu.split("\\|")
          val name: String = stuSplits(1)
          val clazz: String = stuSplits(4)
          s"$id,$name,$clazz,0"
      }.foreach(println)





  }

}
